package com.hkbigdata.StreamCode;

import com.hkbigdata.entry.MarketingUserBehavior;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Random;

/**
 * @author liuanbo
 * @creat 2023-04-26-20:21
 * @see 2194550857@qq.com
 */
public class Flink04_MarketByChannel {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<MarketingUserBehavior> marketingUserBehaviorDataStreamSource = env.addSource(new MarketByChannel());


        KeyedStream<MarketingUserBehavior, Tuple2<String, String>> marketingUserBehaviorTuple2KeyedStream = marketingUserBehaviorDataStreamSource.keyBy(new KeySelector<MarketingUserBehavior, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(MarketingUserBehavior value) throws Exception {
                return new Tuple2<>(value.getChannel(), value.getBehavior());
            }
        });

        SingleOutputStreamOperator<Tuple2<Tuple2<String, String>, Integer>> process = marketingUserBehaviorTuple2KeyedStream.process(new KeyedProcessFunction<Tuple2<String, String>, MarketingUserBehavior, Tuple2<Tuple2<String, String>, Integer>>() {
            private int count = 0;
            private HashMap<String, Integer> hashMap = new HashMap<>();

            @Override
            public void processElement(MarketingUserBehavior value, Context ctx, Collector<Tuple2<Tuple2<String, String>, Integer>> out) throws Exception {
                //1.拼接key
                String hashkey = value.getChannel() + "-" + value.getBehavior();
                //给hashmap赋值
                hashMap.getOrDefault(hashkey, 0);
                count++;
                out.collect(new Tuple2<>(ctx.getCurrentKey(), count));

                hashMap.put(hashkey, count);

            }
        });


        process.print();

        env.execute();


    }

    public static class MarketByChannel extends RichSourceFunction<MarketingUserBehavior> {

        private Random random = new Random();
        private List<String> channels = Arrays.asList("huawei", "xiaomi", "taobao", "vivo", "oppo", "jingdong");
        private List<String> behaviors = Arrays.asList("download", "install", "updated", "uninstall");
        private Boolean scan = true;

        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
            while (scan) {
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        (long) random.nextInt(100000),
                        behaviors.get(random.nextInt(behaviors.size())),
                        channels.get(random.nextInt(channels.size())),
                        System.currentTimeMillis()

                );
                ctx.collect(marketingUserBehavior);
                Thread.sleep(1000);

            }

        }

        @Override
        public void cancel() {

        }
    }
}
